/**
 * 
 */
package tweet;

import java.io.IOException;
import java.text.DateFormat;
import java.text.ParseException;
import java.text.SimpleDateFormat;
import java.util.ArrayList;
import java.util.Calendar;
import java.util.Collections;
import java.util.Comparator;
import java.util.Date;
import java.util.HashMap;
import java.util.Hashtable;
import java.util.List;
import java.util.Map;
import java.util.Map.Entry;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.BinaryComparable;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.Reducer;
import org.apache.hadoop.mapreduce.Reducer.Context;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
import org.apache.hadoop.util.GenericOptionsParser;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.util.GenericOptionsParser;

import tweet.Util;

/**
 * @author vaibhav
 *
 */
public class VolumeTweet {

	/**
	 * @param args
	 */
	
	public static class VolumeTweetMapper extends Mapper<Object, Text, Text, IntWritable> {
		public static final String TAB = "\t";
		public static final String TAG = "#";
		
		
		private HashMap<Integer, String> toptweets = new HashMap<Integer, String>();
		
	
		protected void setup(org.apache.hadoop.mapreduce.Mapper<Object, Text, Text, IntWritable>.Context context)
        throws IOException, InterruptedException {
	
	toptweets.put(1, "#party");
	toptweets.put(2, "#morning");
	toptweets.put(3, "#north");
	toptweets.put(4, "#east");
	toptweets.put(5, "#college");
}

		private IntWritable one = new IntWritable(1);

		public void map(Object key, Text value, Context context) throws IOException, InterruptedException {
			String[] line = value.toString().split(TAB);
			if (line.length > 9) {
				String timestamp = line[2];
				String tweetTxt = line[9];

				if ((tweetTxt.indexOf(TAG) != -1) && (timestamp.indexOf(":") != -1)
				        && (timestamp.split(" ").length == 6)) {
					String tag = tweetTxt.substring(tweetTxt.indexOf(TAG));
					if (tag.indexOf(" ") != -1) {
						tag = tag.substring(0, tag.indexOf(" "));
					}
					
					for(int i=1;i<6;i++)
					{
					
						if (tag.equals(toptweets.get(i)))
						{
							Date postDate = null;
							try {
								postDate = Util.createDate(timestamp);
							} catch (ParseException e) {
								// TODO Auto-generated catch block
								e.printStackTrace();
							}
							Calendar cal = Util.datetocal(postDate);
							
							if(cal.get(Calendar.YEAR)== 109)
							{
								int postmonth = cal.get(Calendar.MONTH); 
								Text ckey = new Text();
								ckey.set(tag+ "-" + postmonth);
								
								// emit(tag g, {t1*t2})
								context.write(ckey, one);
							}
							
							
						}
					}
					
					
				}
			}
		}
	}
	
	public static class TagFindCombiner extends Reducer<Text, IntWritable, Text, IntWritable> {
		public void reduce(Text key, Iterable<IntWritable> values, Context context)
		{
			
			
			int count= 0;


			try {

				for (IntWritable val : values) {

					count = count + val.get();

				}

				IntWritable result = new IntWritable(count);

				context.write(key, result);


			} catch (Exception e) {
				e.printStackTrace();
			}

		}

			
		}
	
	public static class VolumeTweetReducer extends Reducer<Text, IntWritable, Text, IntWritable> {

		public void reduce(Text key, Iterable<IntWritable> values, Context context)
		throws IOException, InterruptedException {

			int count= 0;



			try {

				for (IntWritable val : values) {

					count = count + val.get();

				}

				IntWritable result = new IntWritable(count);

				context.write(key, result);


			} catch (Exception e) {
				// TODO: handle exception
			}

		}
	}

		


	
	public static void main(String[] args) throws IOException, InterruptedException, ClassNotFoundException {
		// TODO Auto-generated method stub

		Configuration conf = new Configuration();
		String[] otherArgs = new GenericOptionsParser(args).getRemainingArgs();
		if (otherArgs.length != 2) {
			System.err.println("Usage: com.neu.mapreduce.tweet.TagFinder <in> <out>");
			System.exit(2);
		}
		Job job = new Job(conf, "Tweet Volume/month");
		job.setJarByClass(TagFinder.class);
		job.setMapperClass(VolumeTweetMapper.class);
		//job.setCombinerClass(TagFindCombiner.class);
		job.setReducerClass(VolumeTweetReducer.class);
		job.setOutputKeyClass(Text.class);
		job.setOutputValueClass(IntWritable.class);
		FileInputFormat.addInputPath(job, new Path(otherArgs[0]));
		FileOutputFormat.setOutputPath(job, new Path(otherArgs[1]));
		System.exit(job.waitForCompletion(true) ? 0 : 1);

	}

}
